package com.kingsoft.dc.khaos.module.spark.util

import java.util

import com.kingsoft.dc.khaos.module.spark.constants.RedisConstants
import com.kingsoft.dc.khaos.module.spark.constants.RedisConstants.RedisRoleEnum
import com.kingsoft.dc.khaos.module.spark.sink.SinkStrategy
import com.kingsoft.dc.khaos.module.spark.source.SourceStrategy
import com.kingsoft.dc.khaos.util.Logging
import com.redislabs.provider.redis.ReadWriteConfig
import com.redislabs.provider.redis.rdd.Keys
import org.apache.commons.lang3.StringUtils
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import redis.clients.jedis._

import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
import scala.util.control.Breaks._

/**
  * Created by haorenhui on 2020/06/28.
  */
object RedisUtils extends Logging{

    private val db: Int = 0
    private val clientName: String = "client"

    /** redis单机模式 conn */
    def getSinglePool[T](redisNodes:String,redisPassword:String,redisProperties:Map[String, String],classType: T): JedisPool = {
        log.info("redis 初始化主从模式连接 ")
        val (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts): (Int, Int, Int, Int, Int, Boolean, Int) = if(classType.isInstanceOf[SourceStrategy]){
            getSourceProperties(redisProperties)
        }else if(classType.isInstanceOf[SinkStrategy]) {
            getSinkProperties(redisProperties)
        }else{
            getDeleteProperties(redisProperties)
        }

        log.info(s"redis connectTimeout: $connectTimeout soTimeout: $soTimeout maxTotal:$maxTotal maxIdle:$maxIdle minIdle:$minIdle testOnBorrow:$testOnBorrow")

        @transient val jedisPoolConfig = new JedisPoolConfig()
        jedisPoolConfig.setMaxTotal(maxTotal)
        jedisPoolConfig.setMaxIdle(maxIdle)
        jedisPoolConfig.setMinIdle(minIdle)
        jedisPoolConfig.setTestOnBorrow(testOnBorrow)
        val hostAndPort: String = redisNodes.split(",")(0)
        if(!hostAndPort.contains(":")){
            throw new IllegalArgumentException(s"redis地址格式错误: $hostAndPort")
        }
        val hostPortArray: Array[String] = hostAndPort.split(":")
        val host: String = hostPortArray(0)
        var port: Int = 6379
        try{
            port = hostPortArray(1).toInt
        }catch {
            case e:NumberFormatException =>
                throw new IllegalArgumentException(s"redis地址格式错误: $hostAndPort",e)
        }
        lazy val pool :JedisPool= new JedisPool(jedisPoolConfig, host, port, connectTimeout, soTimeout,redisPassword,db,clientName)

        pool
    }

    /** redis集群模式 conn */
    def getClusterPool[T](redisNodes:String,redisPassword:String,redisProperties:Map[String, String],classType: T): JedisCluster = {
        log.info("redis 初始化集群模式连接 ")
        val (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts): (Int, Int, Int, Int, Int, Boolean, Int) = if(classType.isInstanceOf[SourceStrategy]){
            getSourceProperties(redisProperties)
        }else if(classType.isInstanceOf[SinkStrategy]) {
            getSinkProperties(redisProperties)
        }else{
            getDeleteProperties(redisProperties)
        }
        log.info(s"redis connectTimeout: $connectTimeout soTimeout: $soTimeout maxTotal:$maxTotal maxIdle:$maxIdle minIdle:$minIdle testOnBorrow:$testOnBorrow maxAttempts:$maxAttempts")

        val set = new util.HashSet[HostAndPort]()
        val hostAndPortArr: Array[String] = redisNodes.split(",")
        hostAndPortArr.foreach(line => {
            if(!line.contains(":")){
                throw new IllegalArgumentException(s"redis地址格式错误: $line")
            }
            val array: Array[String] = line.split(":")
            val host: String = array(0)
            var port: Int = 6379
            try{
                port = array(1).toInt
            }catch {
                case e:NumberFormatException =>
                    throw new IllegalArgumentException(s"redis地址格式错误: $line",e)
            }
            val hostAndPort = new HostAndPort(host, port)
            set.add(hostAndPort)
        })

        @transient val poolConfig = new GenericObjectPoolConfig()
        poolConfig.setMaxTotal(maxTotal)
        poolConfig.setMaxIdle(maxIdle)
        poolConfig.setMinIdle(minIdle)
        poolConfig.setTestOnBorrow(testOnBorrow)

        val jedisCluster: JedisCluster = new JedisCluster(set, connectTimeout, soTimeout, maxAttempts, redisPassword, poolConfig)
        jedisCluster
    }

    /** redis 哨兵模式 conn */
    def getSentinelPool[T](redisNodes:String,redisPassword:String,masterName:String,redisProperties:Map[String, String],classType: T): JedisSentinelPool = {
        log.info("redis 初始化哨兵模式连接 ")
        val (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts): (Int, Int, Int, Int, Int, Boolean, Int) = if(classType.isInstanceOf[SourceStrategy]){
            getSourceProperties(redisProperties)
        }else if(classType.isInstanceOf[SinkStrategy]) {
            getSinkProperties(redisProperties)
        }else{
            getDeleteProperties(redisProperties)
        }
        log.info(s"redis connectTimeout: $connectTimeout soTimeout: $soTimeout maxTotal:$maxTotal maxIdle:$maxIdle minIdle:$minIdle testOnBorrow:$testOnBorrow")

        val sentinels = new util.HashSet[String]()
        redisNodes.split(",").foreach(hostAndPort => sentinels.add(hostAndPort))
        @transient val poolConfig = new GenericObjectPoolConfig()
        poolConfig.setMaxTotal(maxTotal)
        poolConfig.setMaxIdle(maxIdle)
        poolConfig.setMinIdle(minIdle)
        poolConfig.setTestOnBorrow(testOnBorrow)
        val pool: JedisSentinelPool = new JedisSentinelPool(masterName, sentinels, poolConfig, connectTimeout, soTimeout,redisPassword,db,clientName)
        pool
    }

    def getTypeKey(conn:JedisCluster,keyPattern:String,value_type:String,redisProperties:Map[String, String]): Array[String] ={
        log.info(s"redis scan keyPattern : $keyPattern")
        var scanCount:Int=1000
        scanCount= redisProperties.getOrElse(RedisConstants.SOURCE_PER_SCAN_NUMS,RedisConstants.DEFAULT_SOURCE_PER_SCAN_NUMS).toInt
        log.info(s"redis scan count : $scanCount")
        // 匹配到所有符合的key, 根据key获取对应类型
        val matchKeys = new ListBuffer[String]()

        val clusterNodes: util.Map[String, JedisPool] = conn.getClusterNodes
        clusterNodes.asScala.foreach(node=>{
            val resource: Jedis = node._2.getResource
            val scan = new ScanParams
            scan.`match`(keyPattern)
            scan.count(scanCount)

            var scanResult: ScanResult[String] = resource.scan("0",scan)
            var cursor: String = scanResult.getCursor
            matchKeys.append(scanResult.getResult.asScala:_*)
            while(cursor.toInt != 0){
                scanResult = resource.scan(cursor,scan)
                cursor = scanResult.getCursor
                matchKeys.append(scanResult.getResult.asScala:_*)
            }
        })


        log.info(s" 开始过滤key ")
        val keys:Array[String] = matchKeys.map(keyName => {
            val keyType: String = conn.`type`(keyName)
            (keyName, keyType)
        }).filter(tp=>tp._2.equals(value_type)).map(_._1).toArray
        log.info(s" key过滤完成 ")
        keys.distinct
    }

    def getTypeKey(conn:Jedis,keyPattern:String,value_type:String,redisProperties:Map[String, String]): Array[String] ={
        log.info(s"redis scan keyPattern : $keyPattern")
        var scanCount:Int=1000
        scanCount= redisProperties.getOrElse(RedisConstants.SOURCE_PER_SCAN_NUMS,RedisConstants.DEFAULT_SOURCE_PER_SCAN_NUMS).toInt
        log.info(s"redis scan count : $scanCount")
        // 匹配到所有符合的key, 根据key获取对应类型
        val matchKeys = new ListBuffer[String]()
        val scan = new ScanParams
        scan.`match`(keyPattern)
        scan.count(scanCount)

        var scanResult: ScanResult[String] = conn.scan("0",scan)
        var cursor: String = scanResult.getCursor
        matchKeys.append(scanResult.getResult.asScala:_*)
        while(cursor.toInt != 0){
            scanResult = conn.scan(cursor,scan)
            cursor = scanResult.getCursor
            matchKeys.append(scanResult.getResult.asScala:_*)
        }
        // 匹配到所有符合的key, 根据key获取对应类型
        log.info(s" 开始过滤key ")
        val keys:Array[String] =Keys.filterKeysByType(conn,matchKeys.toArray,value_type)(new ReadWriteConfig(100,100))
        log.info(s" key过滤完成 ")
        keys.distinct
    }

    //todo  扩展
    def getSourceProperties(redisProperties:Map[String, String]): (Int,Int,Int,Int,Int,Boolean,Int) ={
        val connectTimeout:Int = redisProperties.getOrElse(RedisConstants.SOURCE_CONNECTTIMEOUT,RedisConstants.DEFAULT_SOURCE_CONNECTTIMEOUT).toInt
        val soTimeout:Int = redisProperties.getOrElse(RedisConstants.SOURCE_SOTIMEOUT,RedisConstants.DEFAULT_SOURCE_SOTIMEOUT).toInt
        val maxTotal:Int = redisProperties.getOrElse(RedisConstants.SOURCE_MAXTOTAL,RedisConstants.DEFAULT_SOURCE_MAXTOTAL).toInt
        val maxIdle:Int = redisProperties.getOrElse(RedisConstants.SOURCE_MAXIDLE,RedisConstants.DEFAULT_SOURCE_MAXIDLE).toInt
        val minIdle:Int = redisProperties.getOrElse(RedisConstants.SOURCE_MINIDLE,RedisConstants.DEFAULT_SOURCE_MINIDLE).toInt
        val testOnBorrow:Boolean = redisProperties.getOrElse(RedisConstants.SOURCE_TESTONBORROW,RedisConstants.DEFAULT_SOURCE_TESTONBORROW).toBoolean
        val maxAttempts:Int = redisProperties.getOrElse(RedisConstants.SOURCE_CLUSTER_MAXATTEMPTS,RedisConstants.DEFAULT_SOURCE_CLUSTER_MAXATTEMPTS).toInt
        (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts)
    }

    def getSinkProperties(redisProperties:Map[String, String]): (Int,Int,Int,Int,Int,Boolean,Int) ={
        val connectTimeout:Int = redisProperties.getOrElse(RedisConstants.SINK_CONNECTTIMEOUT,RedisConstants.DEFAULT_SINK_CONNECTTIMEOUT).toInt
        val soTimeout:Int = redisProperties.getOrElse(RedisConstants.SINK_SOTIMEOUT,RedisConstants.DEFAULT_SINK_SOTIMEOUT).toInt
        val maxTotal:Int = redisProperties.getOrElse(RedisConstants.SINK_MAXTOTAL,RedisConstants.DEFAULT_SINK_MAXTOTAL).toInt
        val maxIdle:Int = redisProperties.getOrElse(RedisConstants.SINK_MAXIDLE,RedisConstants.DEFAULT_SINK_MAXIDLE).toInt
        val minIdle:Int = redisProperties.getOrElse(RedisConstants.SINK_MINIDLE,RedisConstants.DEFAULT_SINK_MINIDLE).toInt
        val testOnBorrow:Boolean = redisProperties.getOrElse(RedisConstants.SINK_TESTONBORROW,RedisConstants.DEFAULT_SINK_TESTONBORROW).toBoolean
        val maxAttempts:Int = redisProperties.getOrElse(RedisConstants.SINK_CLUSTER_MAXATTEMPTS,RedisConstants.DEFAULT_SINK_CLUSTER_MAXATTEMPTS).toInt
        (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts)
    }

    def getDeleteProperties(redisProperties:Map[String, String]): (Int,Int,Int,Int,Int,Boolean,Int) ={
        val connectTimeout:Int = redisProperties.getOrElse(RedisConstants.DELETE_CONNECTTIMEOUT,RedisConstants.DEFAULT_DELETE_CONNECTTIMEOUT).toInt
        val soTimeout:Int = redisProperties.getOrElse(RedisConstants.DELETE_SOTIMEOUT,RedisConstants.DEFAULT_DELETE_SOTIMEOUT).toInt
        val maxTotal:Int = redisProperties.getOrElse(RedisConstants.DELETE_MAXTOTAL,RedisConstants.DEFAULT_DELETE_MAXTOTAL).toInt
        val maxIdle:Int = redisProperties.getOrElse(RedisConstants.DELETE_MAXIDLE,RedisConstants.DEFAULT_DELETE_MAXIDLE).toInt
        val minIdle:Int = redisProperties.getOrElse(RedisConstants.DELETE_MINIDLE,RedisConstants.DEFAULT_DELETE_MINIDLE).toInt
        val testOnBorrow:Boolean = redisProperties.getOrElse(RedisConstants.DELETE_TESTONBORROW,RedisConstants.DEFAULT_DELETE_TESTONBORROW).toBoolean
        val maxAttempts:Int = redisProperties.getOrElse(RedisConstants.DELETE_CLUSTER_MAXATTEMPTS,RedisConstants.DEFAULT_DELETE_CLUSTER_MAXATTEMPTS).toInt
        (connectTimeout,soTimeout,maxTotal,maxIdle,minIdle,testOnBorrow,maxAttempts)
    }

    def getJedisProperties(redisProperties:Map[String, String]): (Int,Int) ={
        val connectTimeout:Int = redisProperties.getOrElse(RedisConstants.JEDIS_CONNECTTIMEOUT,RedisConstants.DEFAULT_JEDIS_CONNECTTIMEOUT).toInt
        val soTimeout:Int = redisProperties.getOrElse(RedisConstants.JEDIS_SOTIMEOUT,RedisConstants.DEFAULT_JEDIS_SOTIMEOUT).toInt
        (connectTimeout,soTimeout)
    }

    /**
      * 根据key删除redis中数据
      * */
    def deleteDataByKey(conn:Jedis, key:String): Unit ={
        conn.del(key)
    }

    /**
      * 根据key数组删除redis中数据
      * */
    def deleteDataByKeys(conn:Jedis, keys:Array[String]): Unit ={
        conn.del(keys:_*)
    }

    /**
      * 根据key删除redis中数据
      * */
    def deleteDataByKey(conn:JedisCluster, key:String): Unit ={
        conn.del(key)
    }

    /**
      * 根据key数组删除redis中数据
      * */
    def deleteDataByKeys(conn:JedisCluster, keys:Array[String]): Unit ={
        conn.del(keys:_*)
    }

    /**
      * 关闭redis连接
      * */
    def closeConn(conn:Jedis): Unit ={
        if(null != conn && conn.isConnected)
            conn.close()
        log.info("redis 连接关闭")
    }

    /**
      * 关闭redis集群连接
      * */
    def closeConn(conn:JedisCluster): Unit ={
        if(null != conn)
            conn.close()
        log.info("redis cluster 连接关闭")
    }

    /**
      * 关闭redis连接池
      * */
    def closeConnPool(conn: JedisPoolAbstract): Unit ={
        if(null != conn)
            conn.close()
        log.info("redis 连接池关闭")
    }


    def getMasterSlaveMasterNode(redisNodes:String,password:String,redisProperties:Map[String, String]): String ={
        val (connectTimeout,soTimeout): (Int, Int) = getJedisProperties(redisProperties)
        var masterNode:String = null
        log.info("redis try getMasterSlaveMasterNode")
        if(StringUtils.isBlank(redisNodes)){
            throw new Exception(s"redisNodes is empty : $redisNodes")
        }
        val nodes: Array[String] = redisNodes.split("\\,")
        var masterAvailable: Boolean = false

        for (node <- nodes) {
            if(!masterAvailable){
                var jedis:Jedis = null
                try{
                    val hostAndPort: HostAndPort = HostAndPort.parseString(node)
                    jedis = new Jedis(hostAndPort.getHost, hostAndPort.getPort, connectTimeout, soTimeout)

                    if(password!=null){
                        jedis.auth(password)
                    }
                    val replication: String = jedis.info("Replication")
                    val infoMap: Map[String, String] = parseInfo(replication)
                    val role: String = infoMap.getOrElse("role", null)
                    role match {
                        case RedisRoleEnum.master =>
                            log.info(s"redis node $node is master , return node")
                            masterAvailable = true
                            masterNode = node
                        case RedisRoleEnum.slave =>
                            log.info(s"redis node $node is slave, check next")
                            if(!masterAvailable){
                                val master_host: String = infoMap.getOrElse("master_host","")
                                val master_port: String = infoMap.getOrElse("master_port","")
                                masterNode = master_host + ":" + master_port
                            }
                        case _ =>
                            log.info(s"redis node $node is $role, check next")
                    }
                }catch {
                    case e:Exception =>
                        log.warn(s"try getMaster fail,node $node, info: ${e.getLocalizedMessage} , ${e.getCause.toString}")
                }finally {
                    if(null != jedis){
                        jedis.close()
                    }
                }
            }
        }
        if(!masterAvailable && null != masterNode){
            log.warn(s"redis all node is not master, slave get info master is : $masterNode")
        }
        masterNode
    }

    /**
     * 转换jedis.info内容
     * @param info jedis.info
     * @return info-kvMap
     */
    def parseInfo(info:String): Map[String,String] ={
        val infoMap: util.HashMap[String,String] = new util.HashMap[String,String]()
        if(StringUtils.isNotBlank(info)){
            val infoArr: Array[String] = info.split("\n")
            infoArr.foreach((line: String) =>{
                val keyAndValue: Array[String] = line.split(":")
                if(keyAndValue.length == 2 ){
                    val key: String = keyAndValue(0).trim
                    val value: String = keyAndValue(1).trim
                    infoMap.put(key,value)
                }
            })
        }
        infoMap.asScala.toMap
    }

}
